package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.io.IOException;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
import java.util.*;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.http.client.fluent.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author 张明诚
 * @version V1.0
 * @date 2016-11-18 10:39:47
 * @desc
 */
public class DeduplicationFilterForIvr implements FilterInterface {

	private final Logger LOG = LoggerFactory.getLogger(DeduplicationFilterForIvr.class);
	private String[] columnNames;
	private REFieldType[] types;
	private String docSchemaName;
	private Map<String, String> userDesc;
	private List<Object[]> datas;
	private final String[] urls;
	private final List<Object> deduplicationFields;
	private Map<String, Long> ivrMsgCount;
	private Map<String, Long> ctMsg;
	private int idIndex;

	private int wxsfromIndex;
	private int urlIndex;
	private int wxauIndex;
	private int contIndex;
	private int ctIndex;

	String[] ivrCountMsgColumnNames;
	REFieldType[] ivrCountMsgTypes;

	String[] ctMsgColumnNames;
	REFieldType[] ctMsgTypes;

	public DeduplicationFilterForIvr(Configuration configuration) {
		this.urls = configuration.getStringArray(Constants.DEDUPLICATION_URL);
		this.deduplicationFields = configuration.getList(Constants.DEDUPLICATION_FIELDS);
		initIvrCountMsgConf();
		initCtMsgConf();
	}

	@Override
	public REMessage filter(REMessage message) throws Exception {
		columnNames = message.colNames;
		types = message.types;
		docSchemaName = message.docSchemaName;
		userDesc = message.userDesc;
		datas = message.data;
		ivrMsgCount = new HashMap<>();
		ctMsg = new HashMap<>();

		if (datas.isEmpty()) {
			return message;
		}

		String deduplicationValues = getDeduplicationValues();//得到要去重字段的值
		if (deduplicationValues.isEmpty()) {
			throw new Exception("deduplication values is empty!");
		} else {
			LOG.debug("deduplication values: \n{}", deduplicationValues);
		}
		String response = deduplicationService(deduplicationValues);
		LOG.debug("deduplication result:\n{}", response);
		String[] split = response.split("\n");
		List<String> flags = Arrays.asList(split);
		deleteDuplicationDatasForIvrMsg(flags);
		REMessage deduplicationMessage
				= new REMessage(columnNames, types, datas, docSchemaName, userDesc);
		return deduplicationMessage;
	}

	public String getDeduplicationValues() {
		StringBuilder sb = new StringBuilder();
		List<String> columnList = Arrays.asList(columnNames);

		idIndex = columnList.indexOf("_id");
		ctIndex = columnList.indexOf("ct");

		/*wxsfromIndex = columnList.indexOf("wx_sfrom");
        urlIndex = columnList.indexOf("url");
        wxauIndex = columnList.indexOf("wx_au");
        contIndex = columnList.indexOf("cont");
        ctIndex = columnList.indexOf("ct");*/
		for (Object[] data : this.datas) {
			boolean isFirst = true;
			for (Iterator<Object> it = deduplicationFields.iterator(); it.hasNext();) {
				String distinctField = (String) it.next();
				int index = columnList.indexOf(distinctField);
				if (index < 0) {
					LOG.info(distinctField);
					LOG.error("the distinct field: {} does not exists!", distinctField);
				}
				Object distinctFieldValue = data[index];
//                Object distinctFieldValue = getDistinctFieldValue(data);
				if (distinctFieldValue == null || distinctFieldValue.toString().isEmpty()) {
					LOG.info("columnames:{}", Arrays.toString(columnNames));
					LOG.info("data:{}", Arrays.toString(data));
					LOG.error("the value of distinct field: {} is null or empty!", distinctField);
				}
				if (!isFirst) {
					sb.append("^");
				}
				sb.append(distinctFieldValue);
				isFirst = false;
			}
			sb.append("\n");
		}
		return sb.toString();
	}

	private String getDistinctFieldValue(Object[] data) {

		SimpleDateFormat sf = new SimpleDateFormat("yyyyMMdd");
		StringBuffer sb = new StringBuffer();
		try {
			String ct2day = sf.format(new Date(((Long) data[ctIndex]) * 1000));//ct字段转为天
			sb.append(ct2day)
					.append(",")
					.append(data[wxsfromIndex].toString())
					.append(",")
					.append(data[urlIndex].toString())
					.append(",");
			for (Object obj : (Object[]) data[wxauIndex]) {
				LOG.debug("wxau:{}, obj:{}", data[wxauIndex], obj);
				sb.append(obj.toString())
						.append(",");
			}
			sb.append(data[contIndex].toString());

		} catch (Exception e) {
			LOG.error("getDistinctFieldValue:{} error:{}", sb, e);
		}

		String afterMd5 = string2MD5(sb.toString());
		LOG.debug("wx_sfrom:{},md5:{}", data[wxsfromIndex].toString(), afterMd5);
		LOG.debug("url:{},md5:{}", data[urlIndex].toString(), afterMd5);
		LOG.debug("wxau:{},md5:{}", data[wxauIndex], afterMd5);
		LOG.debug("cont:{},md5:{}", data[contIndex].toString(), afterMd5);

		return afterMd5;
	}

	public void deleteDuplicationDatasForIvrMsg(List<String> flags) {
		List<Object[]> newMessage = new ArrayList<>();
		for (int i = 0; i < flags.size(); i++) {
			//redis不存在的flag为 1
			if (flags.get(i).equals("1")) {
				newMessage.add(datas.get(i));
			}
//            ivrMsgCount.put(String.valueOf(datas.get(i)[idIndex]), Long.valueOf(flags.get(i)));
			//只要去重，就发1
			ivrMsgCount.put(String.valueOf(datas.get(i)[idIndex]), 1L);
			ctMsg.put(String.valueOf(datas.get(i)[idIndex]), (Long) datas.get(i)[ctIndex]);
		}
//        LOG.info("ivrMsgCount :{}", ivrMsgCount.toString());
		LOG.info("ctMsg :{}", ctMsg.toString());
		this.datas = newMessage;
	}

	private String deduplicationService(String content) {
		if (LOG.isDebugEnabled()) {
			LOG.debug("deduplication content:{}", content);
		}
		String response = null;
		String url;
		int index;
		while (true) {
			index = new Random().nextInt(urls.length);
			url = urls[index];
			try {
				response = Request.Post(url)
						.bodyByteArray(content.getBytes())
						.connectTimeout(1800000)
						.socketTimeout(1800000)
						.execute()
						.returnContent().asString();
			} catch (IOException ex) {
				LOG.warn("request url {} error! \n"
						+ ExceptionUtils.getFullStackTrace(ex), url);
			}
			if (response == null || response.trim().isEmpty()) {
				LOG.error("response content is null，please check the url:{}", url);
			} else {
				break;
			}
			try {
				Thread.sleep(3000);
			} catch (InterruptedException ex) {
			}
		}
		LOG.info("deduplication fields:{}, response: {} ", content, response);
		return response;
	}

	public REMessage getCtMsg() {
		List<Object[]> ctMsgDatas = new ArrayList<>();
		for (Map.Entry<String, Long> entrySet : ctMsg.entrySet()) {
			ctMsgDatas.add(new Object[]{entrySet.getKey(), entrySet.getValue()});
		}
		REMessage msg = new REMessage(ctMsgColumnNames, ctMsgTypes,
				ctMsgDatas, null, null);
		return msg;
	}

	public REMessage getIvrCountMsg() {
		List<Object[]> ivrCountMsgDatas = new ArrayList<>();
		for (Map.Entry<String, Long> entrySet : ivrMsgCount.entrySet()) {
			ivrCountMsgDatas.add(new Object[]{entrySet.getKey(), entrySet.getValue()});
		}
		REMessage msg = new REMessage(ivrCountMsgColumnNames, ivrCountMsgTypes,
				ivrCountMsgDatas, null, null);

		return msg;
	}

	private void initIvrCountMsgConf() {
		ivrCountMsgColumnNames = new String[]{"_id", "count"};
		ivrCountMsgTypes = new REFieldType[]{REFieldType.String, REFieldType.Long};
	}

	private void initCtMsgConf() {
		ctMsgColumnNames = new String[]{"_id", "ct"};
		ctMsgTypes = new REFieldType[]{REFieldType.String, REFieldType.Long};
	}

	public String string2MD5(String plaintext) {
		MessageDigest m;
		try {
			m = MessageDigest.getInstance("MD5");
		} catch (NoSuchAlgorithmException e) {
			System.out.println(e.toString());
			e.printStackTrace();
			return "";
		}
		m.reset();
		m.update(plaintext.getBytes());
		byte[] digest = m.digest();
		BigInteger bigInt = new BigInteger(1, digest);
		String hashtext = bigInt.toString(16);
		while (hashtext.length() < 32) {
			hashtext = "0" + hashtext;
		}
		return hashtext;
	}
}
